package ma.zhiqiang.dataline.connector.source;

import lombok.extern.slf4j.Slf4j;
import ma.zhiqiang.dataline.core.convert.RecordConvertCsv;
import ma.zhiqiang.dataline.core.convert.RecordConvertJson;
import ma.zhiqiang.dataline.core.dto.Schema;
import ma.zhiqiang.dataline.core.dto.SourceRecord;
import ma.zhiqiang.dataline.core.engine.Source;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.LineIterator;

import java.io.File;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;

@Slf4j
public class FileSource implements Source {
    Properties properties;
    String path;                //file文件
    String format;              //文件格式，csv、json、text
    String encoding;            //文件编码
    LineIterator it;            //行集合

    public FileSource(Properties properties) {
        this.properties = properties;
        this.path = properties.getProperty("path","");
        this.encoding = properties.getProperty("encoding","UTF-8");
        this.format = properties.getProperty("format","text");
    }

    @Override
    public Iterator read() throws Exception {
        try {
            it = FileUtils.lineIterator(new File(this.path),this.encoding);
        } catch (IOException e) {
            log.error("文件读取失败",e);
            throw new RuntimeException(e);
        }
        return it;
    }

    @Override
    public SourceRecord convert(Object record) {
        SourceRecord sourceRecord;
        switch(format){
            case "csv":
                sourceRecord = new RecordConvertCsv(properties).convert(record);break;
            case "json":
                sourceRecord =  new RecordConvertJson(properties).convert(record);break;
            default:
                sourceRecord =  new SourceRecord((String) record);break;
        }
        return sourceRecord;
    }

    @Override
    public void close() throws Exception {
        it.close();
    }
}
